package cn.linjianhui.demo.flink;

import org.apache.flink.api.common.functions.MapFunction;
import org.apache.flink.api.java.tuple.Tuple;
import org.apache.flink.api.java.tuple.Tuple2;
import org.apache.flink.streaming.api.TimeCharacteristic;
import org.apache.flink.streaming.api.datastream.DataStream;
import org.apache.flink.streaming.api.datastream.SingleOutputStreamOperator;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.functions.AssignerWithPeriodicWatermarks;
import org.apache.flink.streaming.api.functions.windowing.WindowFunction;
import org.apache.flink.streaming.api.watermark.Watermark;
import org.apache.flink.streaming.api.windowing.time.Time;
import org.apache.flink.streaming.api.windowing.windows.TimeWindow;
import org.apache.flink.util.Collector;
import org.apache.flink.util.OutputTag;

import java.text.SimpleDateFormat;
import java.util.ArrayList;
import java.util.Collections;
import java.util.List;
import java.util.stream.Collectors;

public class StreamingWindowWatermark {

    public static void main(String[] args) throws Exception {
        //获取运行环境
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        //设置使用eventtime，默认是使用processtime
        env.setStreamTimeCharacteristic(TimeCharacteristic.EventTime);

//        env.getConfig().setAutoWatermarkInterval(5000);

        //设置并行度为1,默认并行度是当前机器的cpu数量
        env.setParallelism(1);
        //连接socket获取输入的数据
        DataStream<String> text = env.socketTextStream("192.168.8.206", 9010, "\n");

        //解析输入的数据
        DataStream<Tuple2<String, Long>> inputMap = text.map(new MapFunction<String, Tuple2<String, Long>>() {
            @Override
            public Tuple2<String, Long> map(String value) throws Exception {
                String[] arr = value.split(",");
                return new Tuple2<>(arr[0], Long.parseLong(arr[1]));
            }
        });

        SimpleDateFormat sdf = new SimpleDateFormat("yyyy-MM-dd HH:mm:ss.SSS");

        DataStream<Tuple2<String, Long>> waterMarkStream = inputMap.assignTimestampsAndWatermarks(new AssignerWithPeriodicWatermarks<Tuple2<String, Long>>() {

            Long currentMaxTimestamp = 0L;
            final Long maxOutOfOrderTime = 3000L; // 最大允许的乱序时间是3s

            @Override
            public Watermark getCurrentWatermark() {
                return new Watermark(currentMaxTimestamp - maxOutOfOrderTime);
            }

            @Override
            public long extractTimestamp(Tuple2<String, Long> element, long previousElementTimestamp) {
                long timestamp = element.f1;
                currentMaxTimestamp = Math.max(timestamp, currentMaxTimestamp);

                System.out.println(String.format("Key: %s     EventTime: %s    WaterMark:  %s",
                        element.f0, sdf.format(timestamp), sdf.format(getCurrentWatermark().getTimestamp())));

                return timestamp;
            }
        });

        OutputTag<Tuple2<String, Long>> outputTag = new OutputTag<Tuple2<String, Long>>("late"){};
        SingleOutputStreamOperator<String> window = waterMarkStream.keyBy(0)
                .timeWindow(Time.seconds(5L))
                // 允许延迟5s之后才销毁计算过的窗口
                .allowedLateness(Time.seconds(2))
                .sideOutputLateData(outputTag)
                .apply(new WindowFunction<Tuple2<String, Long>, String, Tuple, TimeWindow>() {
                    @Override
                    public void apply(Tuple tuple, TimeWindow window, Iterable<Tuple2<String, Long>> input, Collector<String> out) throws Exception {
                        String key = tuple.toString();
                        List<Long> arrayList= new ArrayList<>();
                        for (Tuple2<String, Long> next : input) {
                            arrayList.add(next.f1);
                        }
                        Collections.sort(arrayList);
                        String result = String.format("Key: %s     StartTime:  %s    EndTime:  %s     Data: %s",
                                key, sdf.format(window.getStart()), sdf.format(window.getEnd()),
                                arrayList.stream().map(sdf::format).collect(Collectors.joining("|")));

                        out.collect(result);
                    }
                });
        //测试-把结果打印到控制台即可
        window.print("window计算结果:");

        DataStream<Tuple2<String, Long>> late = window.getSideOutput(outputTag);
        late.print("迟到的数据：");

        env.execute("StreamingWindowWatermark");
    }
}
